package utils.config;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

import static utils.config.MyConfig.KAFKA_BOOTSTRAP_SERVER;

//设置Flink流处理从kafka中获取数据
public class MyFlinkUtils {
    public static DataStreamSource<String> getKafkaEnv(StreamExecutionEnvironment env, String topic) {

        env.setParallelism(4);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", KAFKA_BOOTSTRAP_SERVER);
        properties.setProperty("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        return env.addSource(new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties).setStartFromEarliest());

    }
}
